package com.lagou.client;

import com.lagou.server.service.UserService;
import com.lagou.server.service.util.ZookeeperUtil;
import org.I0Itec.zkclient.ZkClient;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

/**
 * 客户端启动类
 * @author yzh
 */
public class ClientBootStrap {

    /**
     * 连接池对象
     */
    private static CountDownLatch countDownLatch = new CountDownLatch(1);

    /**
     * zookeeper客户端对象
     */
    private static ZkClient zkClient;

    /**
     * Rpc消费者对象
     */
    private static RpcConsumer rpcConsumer;

    /**
     * 线程集合:将服务调用以线程封装到集合中
     */
    private static Map<String, Thread> threadMap = new ConcurrentHashMap<>();

    public static void main(String[] args) throws InterruptedException {
        //1.声明rpc消费者实体类
        rpcConsumer = new RpcConsumer();
        //2.从zookeeper上读取服务端列表
        zkClient = ZookeeperUtil.getZkClient();
        List<String> serverList = zkClient.getChildren(ZookeeperUtil.ROOT_PATH);
        for (int i = 0; i < serverList.size(); i++) {
            //获得主机根路径
            String hostPath = serverList.get(i);
            if(hostPath.equals("response")) {
                continue;
            }
            //进行主机子列表监听
            zkClient.subscribeChildChanges(ZookeeperUtil.ROOT_PATH+"/"+hostPath, (String s, List<String> list) -> {
                //服务器上线后,将服务加入调用线程集合
                if(list != null && list.size() > 0) {
                    System.out.println("----------server"+hostPath.substring(4)+"---------已上线");
                    beforeCallService(hostPath);
                } else {//服务器下线后,将服务从调用线程集合中移除,并唤醒循环调用服务线程
                    System.out.println("----------server"+hostPath.substring(4)+"---------已下线");
                    threadMap.remove(hostPath);
                    countDownLatch.countDown();
                }
            });
            List<String> children = zkClient.getChildren(ZookeeperUtil.ROOT_PATH + "/" + hostPath);
            if(children != null && children.size() > 0) {
                beforeCallService(hostPath);
            }
        }
        //3.进行服务循环调用
        while(true) {
            try {
                if(threadMap.size() > 0) {
                    if(threadMap.size() == 1) {
                        threadMap.forEach((k, thread) -> {
                            //执行调用服务方法
                            thread.run();
                            try {
                                Thread.sleep(2000);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        });
                    } else {//负载均衡
                        long host1Time = 0L;
                        long host2Time = 0L;
                        try {
                            if(ZookeeperUtil.getZkClient().exists(ZookeeperUtil.ROOT_PATH+"/response/host1/responseTime")) {
                                host1Time = Long.parseLong(zkClient.readData(ZookeeperUtil.ROOT_PATH+"/response/host1/responseTime"));
                            }
                            if(ZookeeperUtil.getZkClient().exists(ZookeeperUtil.ROOT_PATH+"/response/host2/responseTime")) {
                                host2Time = Long.parseLong(zkClient.readData(ZookeeperUtil.ROOT_PATH+"/response/host2/responseTime"));
                            }
                        } catch (Exception e) {
                        }
                        String minHost = "host1";
                        if(host2Time > 0 && host1Time > 0) {
                            if(host2Time < host1Time) {
                                minHost = "host2";
                            }
                        }
                        threadMap.get(minHost).run();
                        Thread.sleep(2000);
                    }
                }
            } catch (Exception e) {//当线程调用异常,既服务下线后,阻塞线程,当服务从调用线程集合中移除后,再进行唤醒
                countDownLatch.await();
            }
        }
    }

    /**
     * 调用服务前置方法
     * @param hostPath
     */
    private static void beforeCallService(String hostPath) {
        String hostname = zkClient.readData(ZookeeperUtil.ROOT_PATH+"/"+hostPath+"/hostname")+"";
        int port = Integer.parseInt(zkClient.readData(ZookeeperUtil.ROOT_PATH+"/"+hostPath+"/port")+"");
        UserService proxy = (UserService) rpcConsumer.createProxy(UserService.class, hostname, port);
        threadMap.put(hostPath, new Thread(() -> {
            System.out.println(proxy.sayHello("are you ok?").get("msg"));
        }));
    }

}
